package SparkGraphXInAction.SemiSupervisedByGraph

import org.apache.spark._
import org.apache.log4j.{Level, Logger}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.SparkContext._
import org.apache.spark.graphx._
import org.apache.spark.graphx.Graph._
import org.apache.spark.rdd.RDD
import org.apache.spark.graphx.util.GraphGenerators
import org.apache.spark.mllib.clustering._

import scala.util.Random
import scala.collection.mutable.HashMap

/**
  * Created by Administrator on 2017/4/28 0028.
  *   （1）我们首先实现了一个K近邻图构建算法（与进行预测的K近邻算法做好区分，他们不是一样的算法，本书也不会涉及K近邻算法），作为我们的无监督学
  *    习算法，并将它应用于大部分是无标数据的数据集上。
  *   （2）然后我们实现了一个简单的标签传播算法，它会将标签传递给附近的无标顶点上。
  *   （3）最后，我们实现了一个简单的knnPredict()函数来预测新数据点的标签。
  */
object SemiSupervised {

  case class knnVertex(classNum:Option[Int], pos:Array[Double]) extends Serializable{
    //这里dist方法是计算了两个不同的knnVertex的pos的欧氏距离。
    def dist(that:knnVertex) = math.sqrt(pos.zip(that.pos).map(x =>(x._1-x._2)*(x._1-x._2)).reduce(_ + _))
  }
  // 这段程序内部的逻辑要好好研究一下。
  //a是一个class scala.collection.immutable.Vector。其中每个元素是knnVertex。knnVertex有两个元素组成：第一个元素是knnVertex的
  // 类别（类别1还是0？未知就是None），第二个元素是Array of Double,这个Array长度是2，也就是说第二个元素其实是一个二维平面坐标系的坐标。
  //
  def knnGraph(sc:SparkContext, a:Seq[knnVertex], k:Int)={
    //一开始a是一个由knnVertex组成的Vector。对a做了zipWithIndex操作以后，每个元素都在后方增加了一个索引。
    //a2是list of tuple2类型。a2的每个元素是一个tuple2，tuple2中第一个元素是zipWithIndex产生的索引，tuple2中第二个元素就是knnVertex.
    //在zipWithIndex.map(...)操作后，数据类型仍然是class scala.collection.immutable.Vector，我们将其转换为Array类型。
    val a2 = a.zipWithIndex.map(x => (x._2.toLong, x._1)).toArray
    val v = sc.makeRDD(a2)
    //v1._1是每个点的索引，
    //a2.map(...)中的v2._1是点的类别，v2._2和v1._2都是点的二维坐标，dist操作对两个二维坐标求了欧氏距离。
    val e = v.map(v1 => (v1._1, a2.map( v2 => (v2._1, v1._2.dist(v2._2)))
                                  .sortWith((e,f) => e._2 < f._2)//a2.map(...)之后，e,f中的第二个元素是其与v1的欧氏距离，第一个元素是点的类别。sortwith，是将各个点按照与v1的距离，从小到大排序
                                  .slice(1,k+1)//slice这里不懂啊？？？？？？？？？？？？？？？
                                  .map(_._1)))
              .flatMap(x => x._2.map(vid2 =>
              Edge(x._1, vid2, 1/(1+a2(vid2.toInt)._2.dist(a2(x._1.toInt)._2)))))
    Graph(v,e)
  }

  //分布式近似求解的K近邻图生成
  def knnGraphApprox(sc:SparkContext, a:Seq[knnVertex], k:Int)={
    //一开始a是一个由knnVertex组成的Vector。对a做了zipWithIndex操作以后，每个元素都在后方增加了一个索引。
    //a2是list of tuple2类型。a2的每个元素是一个tuple2，tuple2中第一个元素是zipWithIndex产生的索引，tuple2中第二个元素就是knnVertex.
    //在zipWithIndex.map(...)操作后，数据类型仍然是class scala.collection.immutable.Vector，我们将其转换为Array类型。
    val a2 = a.zipWithIndex.map(x => (x._2.toLong, x._1)).toArray
    val v = sc.makeRDD(a2)
    val n = 3
    //v通过下面两行的map和reduce操作，求出了所有点的最大最小x值和最大最小y值。
    val minMax = v.map(x => (x._2.pos(0), x._2.pos(0), x._2.pos(1), x._2.pos(1)))
                  .reduce((a,b) => (math.min(a._1,b._1), math.max(a._2,b._2), math.min(a._3,b._3), math.max(a._4,b._4)))
    //最大最小x值之间的范围
    val xRange = minMax._2 - minMax._1
    //最大最小y值之间的范围
    val yRange = minMax._4 - minMax._3

    //下一行的v.map(...）作用是根据每个点的坐标（x,y）及最大最小x、y以及xRange、yRange来给每个坐标一个区域的名字，每个区域是一个小格，总共有n*n个小格。结果
    //rdd中的每个元素，都是一个tuple2,tuple2中第一个成员是该元素所属的区域（小格），tuple2中第二个成员是原本的x（x就是a2的组成元素）
    def calcEdges(offset: Double) = v.map(x => (math.floor((x._2.pos(0)-minMax._1)/xRange*(n-1)+offset)*n + math.floor((x._2.pos(1) - minMax._3)/yRange *(n-1)+offset),x))
                                      .groupByKey(n*n)//分成n*n个partition
                                      .mapPartitions(ap => {
                                        val af = ap.flatMap(_._2).toList//flatMap()中的_._2是a2的组成元素（zipWithIndex产生的索引，knnVertex）
                                        af.map(v1 => (v1._1, af.map(v2 => (v2._1, v1._2.dist(v2._2)))
                                          .toArray
                                          .sortWith((e,f) =>e._2 < f._2)//根据距离大小来排序。
                                          .slice(1,k+1)//可能是只取每个点距离最近的k个点?????
                                          .map(_._1)))//舍弃距离？？？
                                        .flatMap(x => x._2.map(vid2 =>Edge(x._1, vid2, 1/(1+a2(vid2.toInt)._2.dist(a2(x._1.toInt)._2)))))
                                          .iterator
                                      })
    val e = calcEdges(0.0).union(calcEdges(0.5))
                          .distinct
                          .map(x => (x.srcId, x))
                          .groupByKey
                          .map(x => x._2.toArray.sortWith((e,f) => e.attr > f.attr).take(k))
                          .flatMap(x => x)
    Graph(v,e)
  }

  def semiSupervisedLabelPropagation(g:Graph[knnVertex,Double], maxIterations:Int = 0)={
    //规定标签传播算法的最大迭代次数。
    val maxIter = if(maxIterations == 0) g.vertices.count / 2 else maxIterations
    var g2 = g.mapVertices((vid, vd) => (vd.classNum.isDefined, vd))
    var isChanged = true
    var i = 0

    do{
      val newV = g2.aggregateMessages[Tuple2[Option[Int],HashMap[Int,Double]]](
        ctx => {
          ctx.sendToSrc((ctx.srcAttr._2.classNum, if (ctx.dstAttr._2.classNum.isDefined) HashMap(ctx.dstAttr._2.classNum.get -> ctx.attr) else HashMap[Int, Double]()))
          if (ctx.srcAttr._2.classNum.isDefined)
            ctx.sendToDst((None, HashMap(ctx.srcAttr._2.classNum.get -> ctx.attr)))
        },
        (a1, a2) => {
          if (a1._1.isDefined)
            (a1._1, HashMap[Int, Double]())
          else if (a2._1.isDefined)
            (a2._1, HashMap[Int,Double]())
          else
            (None, a1._2 ++ a2._2.map{ case (k,v) => k -> (v+a1._2.getOrElse(k,0.0)) })
        }
      )

      val newVClassVoted = newV.map(x => (x._1,
            if(x._2._1.isDefined)
              x._2._1
            else if (x._2._2.size > 0 )
              Some(x._2._2.toArray.sortWith((a,b) => a._2 > b._2)(0)._1)
            else None
      ))

      isChanged = g2.vertices.join(newVClassVoted)
                              .map(x => x._2._1._2.classNum != x._2._2)
                              .reduce(_ || _)

      g2 = g2.joinVertices(newVClassVoted)((vid, vd1, u ) =>
        (vd1._1, knnVertex(u, vd1._2.pos)))
      i += 1
    } while(i < maxIter && isChanged)

    g2.mapVertices((vid,vd) => vd._2)
  }

  //基于半监督学习所得的图的预测函数
  def knnPredict[E](g:Graph[knnVertex,E],pos:Array[Double]) =
    g.vertices
      .filter(_._2.classNum.isDefined)
      .map(x => (x._2.classNum.get, x._2.dist(knnVertex(None,pos))))
      .min()(new Ordering[Tuple2[Int,Double]]{
        override def compare(a:Tuple2[Int, Double], b:Tuple2[Int, Double]): Int = a._2.compare(b._2)
      })
      ._1

  def main(args: Array[String]): Unit = {
    // 屏蔽日志
    Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
    Logger.getLogger("org.eclipse.jetty.server").setLevel(Level.OFF)

    //设置运行环境
    val conf = new SparkConf().setAppName("SimpleGraphX").setMaster("local")
    val sc = new SparkContext(conf)
    //生成样本数据
    Random.setSeed(17L)
    val n = 10
    //以下是定义a
    //(1 to n*2)生成1到20的排列
    val a = (1 to n*2).map(i => {
      val x = Random.nextDouble();//生成0到1的随机数。
      if(i <= n)
        knnVertex(if (i%n == 0) Some(0) else None, Array(x*50,20+(math.sin(x*math.Pi) + Random.nextDouble/2)*25))
      else
        knnVertex(if (i%n == 0) Some(1) else None, Array(x*50 + 25, 30 - (math.sin(x*math.Pi) + Random.nextDouble/2)*25))
    })
    //以上是定义a
    //g是一个Graph，它的每个顶点是一个tuple2，tuple2中的第一个元素是顶点的ID，第二个值是knnVertex.每个边的属性是边所相连的两个顶点之间的欧氏距离。
    //g中的4个边三元组（triplets）示例如下：
    /*
    ((13,knnVertex(None,[D@74971ed9)),(15,knnVertex(None,[D@29fc1a2b)),0.06723230848771827)
    ((13,knnVertex(None,[D@74971ed9)),(16,knnVertex(None,[D@2571066a)),0.07000303325368362)
    ((14,knnVertex(None,[D@4455f57d)),(10,knnVertex(None,[D@878537d)),0.14411298384025592)
    ((14,knnVertex(None,[D@4455f57d)),(15,knnVertex(None,[D@29fc1a2b)),0.0975063331945464)
     */
    val g = knnGraph(sc, a, 4)
    //运行半监督学习标签传播算法
    val gs = semiSupervisedLabelPropagation(g)
    val the_class = knnPredict(gs, Array(30.0, 30.0))
    println(the_class)
  }
}
